package com.shiguiwu.springboot3.rockectmq.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @description: 消费者
 * @author: stone
 * @date: Created by 2023/11/10 20:49
 * @version: 1.0.0
 * @pakeage: com.shiguiwu.springboot3.message.rockectmq.consumer
 */
@Slf4j
public class ConsumerTests {

    public final static String consumerGroup = "consumerGroup11";
    public final static String TOPIC_MESSAGE = "string-topic";
    public final static String nameServer = "192.168.192.130:9876";


    /**
     * 服务端推送消息
     */
    @Test
    public void testPushMessage() throws Exception {
        // 初始化consumer，并设置consumer group name
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);

        // 设置NameServer地址
        consumer.setNamesrvAddr(nameServer);
        //订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息
        consumer.subscribe(TOPIC_MESSAGE, "*");
        //注册回调接口来处理从Broker中收到的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                log.warn("线程={} Receive New Messages: {}", Thread.currentThread().getName(), msgs);
                // 返回消息消费状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动Consumer
        consumer.start();
        log.info("Consumer Started.%n");

    }

    /**
     * 服务端推送消息,消息按顺序消费
     */
    @Test
    public void testPushOrderMessage() throws Exception {
        // 初始化consumer，并设置consumer group name
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);

        // 设置NameServer地址
        consumer.setNamesrvAddr(nameServer);
        //订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息
        consumer.subscribe(TOPIC_MESSAGE, "*");
        //注册回调接口来处理从Broker中收到的消息

        consumer.registerMessageListener(new MessageListenerOrderly() {
            final AtomicLong consumeTimes = new AtomicLong(0);
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                log.warn("线程={} Receive New Messages: {}", Thread.currentThread().getName(), msgs);

                this.consumeTimes.incrementAndGet();
                if ((this.consumeTimes.get() % 2) == 0) {
                    log.warn("success Receive New Messages: {}",  consumeTimes.incrementAndGet());
                    return ConsumeOrderlyStatus.SUCCESS;
                } else if ((this.consumeTimes.get() % 5) == 0) {
                    context.setSuspendCurrentQueueTimeMillis(3000);
                    log.warn("fail Receive New Messages: {}",  consumeTimes.incrementAndGet());

                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        // 启动Consumer
        consumer.start();
        log.info("Consumer Started.%n");
    }

    /**
     * 传统的pull模式
     */
    @Test
    public void testPullConsumer() throws Exception{
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameServer);
        consumer.start();
        try {
            MessageQueue mq = new MessageQueue();
            mq.setQueueId(0);
            mq.setTopic(TOPIC_MESSAGE);
            mq.setBrokerName("broker-a");
            long offset = 26;
            PullResult pullResult = consumer.pull(mq, "*", offset, 32);
            log.info("拉去的结果 ==> {} ==={}", pullResult.getPullStatus(), pullResult.getMsgFoundList());
            if (pullResult.getPullStatus().equals(PullStatus.FOUND)) {
                //System.out.printf("%s%n", pullResult.getMsgFoundList());
                consumer.updateConsumeOffset(mq, pullResult.getNextBeginOffset());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        consumer.shutdown();
    }

    /**
     *  Lite  pull模式
     *  Subscribe模式
     */
    @Test
    public void testLiteSubscribePullConsumer() throws Exception{
        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer(consumerGroup);
        litePullConsumer.subscribe(TOPIC_MESSAGE, "*");
        litePullConsumer.setPullBatchSize(20);
        litePullConsumer.setNamesrvAddr(nameServer);

        litePullConsumer.start();
        try {
            while (litePullConsumer.isRunning()) {
                List<MessageExt> messageExts = litePullConsumer.poll();
                log.info("拉去的结果 ==> {} ", messageExts);

            }
        } finally {
            litePullConsumer.shutdown();
        }
    }


    public static volatile boolean running = true;

    /**
     *  Lite  pull模式
     *  Assign
     *
     *  报错，提示消费者没有开启
     */
    @Test
    public void testLiteAssignPullConsumer() throws Exception{
        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer(consumerGroup);
        litePullConsumer.setNamesrvAddr(nameServer);
        litePullConsumer.setAutoCommit(false);
        litePullConsumer.start();
        Collection<MessageQueue> mqSet = litePullConsumer.fetchMessageQueues(TOPIC_MESSAGE);
        List<MessageQueue> list = new ArrayList<>(mqSet);
        List<MessageQueue> assignList = new ArrayList<>();
        for (int i = 0; i < list.size() / 2; i++) {
            assignList.add(list.get(i));
        }
        litePullConsumer.assign(assignList);
        litePullConsumer.seek(assignList.get(0), 10);
        try {
            while (running) {
                List<MessageExt> messageExts = litePullConsumer.poll();
                log.info("拉去的结果 ==> {} ", messageExts);
                litePullConsumer.commitSync();
            }
        } finally {
            litePullConsumer.shutdown();
        }
    }
}
